package com.we.doris;

import org.apache.doris.flink.cfg.DorisExecutionOptions;
import org.apache.doris.flink.cfg.DorisOptions;
import org.apache.doris.flink.cfg.DorisReadOptions;
import org.apache.doris.flink.sink.DorisSink;
import org.apache.doris.flink.sink.writer.SimpleStringSerializer;

import java.util.Properties;

public class SinkDoris {
    /**
     * 不从配置文件读取,自己传入参数 构建 {@link DorisSink <String>}
     *
     * @param fenodes fe nodes
     * @param tableIdentifier doris 表名
     * @param username 用户名
     * @param password 密码
     * @param sequenceCol sequence col 列
     * @return {@link DorisSink<String>}
     */
    public static DorisSink<String> addDorisSink(
            String fenodes,
            String tableIdentifier,
            String label,
            String username,
            String password,
            String sequenceCol) {
        return buidJsonDorisSink(fenodes, tableIdentifier, label, username, password, sequenceCol);
    }

    /**
     * 构建 {@link DorisSink<String>}
     *
     * @param fenodes fe nodes
     * @param tableIdentifier 表名
     * @param username 用户名
     * @param password 密码
     * @param sequenceCol sequence col 列
     * @return {@link DorisSink<String>}
     */
    public static DorisSink<String> buidJsonArrayDorisSink(
            String fenodes,
            String tableIdentifier,
            String label,
            String username,
            String password,
            String sequenceCol) {
        DorisSink.Builder<String> builder = DorisSink.builder();
        DorisOptions.Builder dorisBuilder =
                DorisOptions.builder()
                        .setFenodes(fenodes)
                        .setTableIdentifier(tableIdentifier)
                        .setUsername(username)
                        .setPassword(password);

        Properties properties = new Properties();
        properties.setProperty("format", "json");
        properties.setProperty("read_json_by_line", "true");
        properties.setProperty("strip_outer_array", "true");
        properties.setProperty("function_column.sequence_col", sequenceCol);

        DorisExecutionOptions.Builder executionBuilder =
                DorisExecutionOptions.builder()
                        // streamload label prefix
                        .setLabelPrefix(label+"_" + System.currentTimeMillis())
                        .setMaxRetries(10)
                        .setStreamLoadProp(properties);
        return builder.setDorisReadOptions(DorisReadOptions.builder().build())
                .setDorisExecutionOptions(executionBuilder.build())
                // serialize according to string
                .setSerializer(new SimpleStringSerializer())
                .setDorisOptions(dorisBuilder.build())
                .build();
    }

    public static DorisSink<String> buidJsonDorisSink(
            String fenodes,
            String tableIdentifier,
            String label,
            String username,
            String password,
            String sequenceCol) {
        DorisSink.Builder<String> builder = DorisSink.builder();
        DorisOptions.Builder dorisBuilder =
                DorisOptions.builder()
                        .setFenodes(fenodes)
                        .setTableIdentifier(tableIdentifier)
                        .setUsername(username)
                        .setPassword(password);

        Properties properties = new Properties();
        properties.setProperty("format", "json");
        properties.setProperty("read_json_by_line", "true");
        properties.setProperty("function_column.sequence_col", sequenceCol);

        DorisExecutionOptions.Builder executionBuilder =
                DorisExecutionOptions.builder()
                        // streamload label prefix
                        .setLabelPrefix(label+"_" + System.currentTimeMillis())
                        .setMaxRetries(10)
                        .setStreamLoadProp(properties);
        return builder.setDorisReadOptions(DorisReadOptions.builder().build())
                .setDorisExecutionOptions(executionBuilder.build())
                // serialize according to string
                .setSerializer(new SimpleStringSerializer())
                .setDorisOptions(dorisBuilder.build())
                .build();
    }
}
